package com.hitqz.robot.dispatch.netty;

import com.hitqz.robot.api.business.entity.SensorModulesEntity;
import com.hitqz.robot.biz.service.SensorModulesService;
import com.hitqz.robot.dispatch.netty.decoder.Server2RobotDecoder;
import com.hitqz.robot.dispatch.netty.dispatch.RobotDispatch;
import com.hitqz.robot.dispatch.netty.encoder.Server2RobotEncoder;
import com.hitqz.robot.dispatch.netty.model.IOReadModulesDto;
import com.hitqz.robot.dispatch.netty.thread.IOModulesReadBatch;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.compress.utils.Lists;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author xupkun
 * @date 2023/7/14
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class NettyServer implements CommandLineRunner {

	private final SensorModulesService sensorModulesService;

	public static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10,20,
			1L, TimeUnit.SECONDS,
			new LinkedBlockingQueue<Runnable>(20),
			Executors.defaultThreadFactory(),
			new ThreadPoolExecutor.DiscardPolicy());

	@Override
	public void run(String... args) throws Exception {
		CompletableFuture.runAsync(()-> {
			try {
				Integer port = 9874;
				log.info("启动服务端端口:" + port);
				NettyServer.bind(port);

			} catch (Exception e) {
				log.info("bind() failed" + e.getMessage());
				throw new RuntimeException(e);
			}
		});

		
		// 传感器模块
		List<SensorModulesEntity> sensorModulesList = sensorModulesService.list();

		List<IOReadModulesDto> ioReadModulesDtos = Lists.newArrayList();

        // 找出ip相同的数据
        Map<String, List<SensorModulesEntity>> ipGroupedModules = sensorModulesList.stream()
                .collect(Collectors.groupingBy(SensorModulesEntity::getIp));

        for (Map.Entry<String, List<SensorModulesEntity>> entry : ipGroupedModules.entrySet()) {
            List<SensorModulesEntity> groupedModules = entry.getValue();
            IOReadModulesDto ioReadModulesDto = new IOReadModulesDto();
			ioReadModulesDto.setIp(groupedModules.get(0).getIp());
            // 假设 IOReadModulesDto 有一个构造函数或方法可以接受 List<SensorModulesEntity>
            ioReadModulesDto.setIoModulesReadList(groupedModules);
            ioReadModulesDtos.add(ioReadModulesDto);
        }

		ioReadModulesDtos.stream().forEach(t-> {
			Thread thread = new Thread(new IOModulesReadBatch(t));
			thread.start();
		});



	}


	@Scheduled(fixedRate = 200L)
	public void dispatch() {
		// 调度线程
		RobotDispatch.dispatch();
	}

	public static void bind(Integer port) {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			log.info("正在启动服务端netty通讯端口成功{}", port );
			ServerBootstrap bootstrap = new ServerBootstrap();
			bootstrap.group(bossGroup, workerGroup)
					.childOption(ChannelOption.TCP_NODELAY, true)
					.channel(NioServerSocketChannel.class)
					.option(ChannelOption.SO_RCVBUF, 1024*1024)
					.option(ChannelOption.RCVBUF_ALLOCATOR,new FixedRecvByteBufAllocator(1024*1024))
					.childHandler(
							new ChannelInitializer<Channel>() {
								@Override
								protected void initChannel(Channel channel) throws Exception {
									ByteBuf delimiter = Unpooled.copiedBuffer(new byte[] {
											(byte) 0xEB, (byte) 0x90
									});
									channel.pipeline().addLast(new DelimiterBasedFrameDecoder(4096*10, delimiter));
									channel.pipeline().addLast(new Server2RobotDecoder());
									channel.pipeline().addLast(new Server2RobotEncoder());
									channel.pipeline().addLast(new NettyMessageHandler());
								}
							}
					);
			// 端口绑定
			ChannelFuture f = bootstrap.bind(port).sync();
			Channel channel = f.channel();
			f.channel().closeFuture().sync();
		} catch (Exception e) {
			log.error("启动 netty 失败:" + e.getMessage());
		} finally {
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}

}
